package org.springService;

import java.util.Properties;

import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

/**
 * Kafka生产
 * @author 张宗朋
 * @date   2018年9月5日
 */
public class KafkaProducerDemo {

	public final static String TOPIC="JAVA_TOPIC";
	private Producer<String,String> createKafkaProducer(){
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhoat:9092,localhoat:9092");
        props.put("acks", "all");
        props.put("retries", 0);
        props.put("batch.size", 16384);
        props.put("linger.ms", 1);
        props.put("buffer.memory", 33554432);
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        Producer<String, String> kafkaProducer = new KafkaProducer<>(props);
        return kafkaProducer;
    }
	
	public static void main(String[] args) {
		KafkaProducerDemo demo = new KafkaProducerDemo();
		Producer<String,String> kafkaProducer = demo.createKafkaProducer();
		String key=String.valueOf("key"+22222);
        String data="测试卡夫卡-------hello kafka message:"+key;
        kafkaProducer.send(new ProducerRecord<>(TOPIC, key, data), new Callback() {
            @Override
            public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                //do sth
            }
        });
	}
}
